package com.xiaomaoguai.fcp.pre.kepler.delay.task.zk;

import com.fasterxml.jackson.annotation.JsonTypeInfo;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.fasterxml.jackson.databind.ser.std.StdSerializer;
import com.sun.xml.internal.ws.encoding.soap.SerializationException;
import com.xiaomaoguai.fcp.pre.kepler.delay.task.DelayTaskMsg;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.springframework.cache.support.NullValue;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import java.io.IOException;

/**
 * 队列数据序列化与反序列化
 *
 * @fileName: DelayTaskMsgQueueSerializer.java
 * @author: WeiHui
 * @date: 2019/2/14 11:01
 * @version: v1.0.0
 * @since JDK 1.8
 */
public class DelayTaskMsgQueueSerializer implements QueueSerializer<DelayTaskMsg> {

	private final ObjectMapper mapper;

	public DelayTaskMsgQueueSerializer() {
		this((String) null);
	}

	public DelayTaskMsgQueueSerializer(@Nullable String classPropertyTypeName) {
		this(new ObjectMapper());
		// simply setting {@code mapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS)} does not help here since we need
		// the type hint embedded for deserialization using the default typing feature.
		mapper.registerModule(new SimpleModule().addSerializer(new NullValueSerializer(classPropertyTypeName)));
		if (StringUtils.hasText(classPropertyTypeName)) {
			mapper.enableDefaultTypingAsProperty(ObjectMapper.DefaultTyping.NON_FINAL, classPropertyTypeName);
		} else {
			mapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
		}
	}

	public DelayTaskMsgQueueSerializer(ObjectMapper mapper) {
		Assert.notNull(mapper, "ObjectMapper must not be null!");
		this.mapper = mapper;
	}

	@Override
	public byte[] serialize(DelayTaskMsg delayTaskMsg) {
		if (delayTaskMsg == null) {
			return new byte[0];
		}
		try {
			return mapper.writeValueAsBytes(delayTaskMsg);
		} catch (JsonProcessingException e) {
			throw new SerializationException("Could not write JSON: " + e.getMessage(), e);
		}
	}

	@Override
	public DelayTaskMsg deserialize(byte[] source) {
		return deserialize(source, DelayTaskMsg.class);
	}

	/**
	 * @param source can be {@literal null}.
	 * @param type   must not be {@literal null}.
	 * @return {@literal null} for empty source.
	 * @throws SerializationException
	 */
	@Nullable
	public <T> T deserialize(@Nullable byte[] source, Class<T> type) throws SerializationException {
		Assert.notNull(type,
				"Deserialization type must not be null! Pleaes provide Object.class to make use of Jackson2 default typing.");
		if (isEmpty(source)) {
			return null;
		}
		try {
			return mapper.readValue(source, type);
		} catch (Exception ex) {
			throw new SerializationException("Could not read JSON: " + ex.getMessage(), ex);
		}
	}

	private class NullValueSerializer extends StdSerializer<NullValue> {

		private static final long serialVersionUID = 1999052150548658808L;

		private final String classIdentifier;

		/**
		 * @param classIdentifier can be {@literal null} and will be defaulted to {@code @class}.
		 */
		NullValueSerializer(@Nullable String classIdentifier) {
			super(NullValue.class);
			this.classIdentifier = StringUtils.hasText(classIdentifier) ? classIdentifier : "@class";
		}

		@Override
		public void serialize(NullValue value, JsonGenerator jgen, SerializerProvider provider)
				throws IOException {
			jgen.writeStartObject();
			jgen.writeStringField(classIdentifier, NullValue.class.getName());
			jgen.writeEndObject();
		}
	}

	private static boolean isEmpty(@Nullable byte[] data) {
		return (data == null || data.length == 0);
	}

}
